package com.yanxu;

import org.apache.commons.validator.Msg;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2023/11/18-2023
 * @Description: 消费者进行消息的消费
 */
public class ConsumerSample01 {
    public static void main(String[] args) {
        //定义消费者  push 字面意思 (推)  。 表示 rocketmq 将消息推送给生产者。
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("yx-consumer");
        consumer.setNamesrvAddr("192.168.1.30:9876");


        try {
            //定义指定的主题
            consumer.subscribe("test-01","*");   // *  代表订阅该主题中的所有消息
            //consumer.subscribe("test-01","test-01");

            //消费模式为集群模式
            consumer.setMessageModel(MessageModel.CLUSTERING);
            //consumer.setMessageModel(MessageModel.BROADCASTING);

            //注册消息监听器。consumeMessage 当收到消息的时候进行消息的处理
            //当有新的消息的时候就会推送过来，触发该 listener 进行消息的消费。
            /**
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    return null;
                }
            });
            */
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    //每一个 MessageExt 就可以简单的理解为一个消息
                    System.err.println("消费线程为：" + Thread.currentThread().getName());
                    System.err.println("消息的个数为：" + list.size());
                    for (MessageExt messageExt : list) {
                        System.out.println("消息的ID为：" + messageExt.getMsgId());
                        System.out.println("消息的内容为：" + new String(messageExt.getBody()));
                        System.out.println("--------------------------------------------------");
                    }

                    //ConsumeConcurrentlyStatus.CONSUME_SUCCESS 表示所有的消息全部消费成功。
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });


            //消费者启动
            consumer.start();
            System.out.println("消费者启动成功:" + Thread.currentThread().getName());

            //挂起当前的jvm
            //System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}